package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.bean.UserRegisterBean;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.MyClickhouseUtil;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
//卢顺祥  用户域用户注册统计
public class DwsUserUserRegisterWindow {
    public static void main(String[] args) throws Exception {

        // TODO 1. 准备环境
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // TODO 2. 读取 Kafka dwd_user_register 主题数据，封装为流
        String topic = "dwd_user_register";
        String groupId = "dws_user_user_register_window";

        // todo 3. 读取数据
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> kafkaDS = env.addSource(kafkaConsumer);

        // todo 4. 转换结构
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(JSON::parseObject);
        // jsonObjDS.print("jsonObjDS>>>>>>");


        // todo 5. 设置水位线
        SingleOutputStreamOperator<JSONObject> withWaterDS = jsonObjDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject jsonObj, long recordTimestamp) {
                                return jsonObj.getLong("ts") * 1000L;
                            }
                        })
        );

        // todo 6. 开窗
        AllWindowedStream<JSONObject, TimeWindow> windowDS =
                withWaterDS.windowAll(TumblingEventTimeWindows.of(Time.seconds(10)));

        // todo 7. 聚合
        SingleOutputStreamOperator<UserRegisterBean> aggregateDS = windowDS.aggregate(

                new AggregateFunction<JSONObject, Long, Long>() {
                    @Override
                    public Long createAccumulator() {
                        return 0L;
                    }

                    @Override
                    public Long add(JSONObject value, Long acc) {
                        return acc += 1L;
                    }

                    @Override
                    public Long getResult(Long accumulator) {
                        return accumulator;
                    }

                    @Override
                    public Long merge(Long a, Long b) {
                        return null;
                    }
                },
                new AllWindowFunction<Long, UserRegisterBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow timeWindow, Iterable<Long> iterable,
                                      Collector<UserRegisterBean> out) throws Exception {
                        for (Long aLong : iterable) {
                            String stt = DateFormatUtil.toYmdHms(timeWindow.getStart());
                            String edt = DateFormatUtil.toYmdHms(timeWindow.getEnd());

                            UserRegisterBean userRegisterBean = new UserRegisterBean(
                                    stt,
                                    edt,
                                    aLong,
                                    System.currentTimeMillis()
                            );
                            out.collect(userRegisterBean);
                        }
                    }
                }
        );

        aggregateDS.print("aggregateDS>>>>>>");

        // todo 8.  写到数据库中

        SinkFunction<UserRegisterBean> userRegisterBeanDataStreamSink = MyClickhouseUtil.<UserRegisterBean>getSinkFunction(
                        "insert into dws_user_user_register_window values(?,?,?,?)" );
        aggregateDS.<UserRegisterBean>addSink(userRegisterBeanDataStreamSink);


        env.execute();
    }
}
